package com.foreveross.fetl.queue;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.annotation.Resource;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.log4j.Logger;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.vo.FetlErrorMessage;
import com.foreveross.fetl.core.domain.doub.DoublePlaneInfoEntity;
import com.foreveross.fetl.core.domain.task.SourceEntity;
import com.foreveross.fetl.queue.util.ExchangeRateUtil;
import com.foreveross.fetl.queue.util.LogMessageUtils;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.FetlErrorUtils;
import com.foreveross.fetl.util.HbaseToPlaneInfoUtil;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.util.StringUtils;

/**
 * 每小时统计一次
 * 
 * @author ljshi
 * 
 */
@Component
public class LastestDoubleDataPreprocess {

	private static Logger log = Logger.getLogger("lastDouble");
	
	
	@Resource(name="taskExecutor")
	private  TaskExecutor taskExecutor;
	
	private int tasksNumber;//任务数 多少条rowkey
	
	private int failuresNumber;//失败数
	
	private int threadsNumber;//线程数目
	
	private int executeNumber;//已执行的任务条数
	
	
	private int insertNumber; //插入条数 
	
	private long excuteStartTime;
	
	
	private String taskTime=null;
	


	public void execute() {
		Date endTime= new Date();
		final Calendar c = Calendar.getInstance();
		c.add(Calendar.HOUR_OF_DAY, -1);
		rollBackNumber();
		taskTime=DateUtils.format(c.getTime(), "yyyy-MM-dd HH");
		process(c.getTime(), endTime);
	}
	
	public void  rollBackNumber(){
		tasksNumber=0;
		failuresNumber=0;
		threadsNumber=0;
		executeNumber=0;
		insertNumber=0;
		excuteStartTime = System.currentTimeMillis();
		LogMessageUtils.create("实时往返价格处理", this.getClass().getSimpleName());

	}
	
	public void manualExecute(final String start,String time){
			this.rollBackNumber();
			try {
			List<String> timeList = Arrays.asList(time.split(";"));
			Calendar c = Calendar.getInstance();
			taskTime = start + " " +time;
			for (String s : timeList) {
				String str =start;
				str += s;
				Date startTime = DateUtils.parse(str, "yyyyMMddHH");
				c.setTime(startTime);
				c.add(Calendar.HOUR_OF_DAY, +1);
				Date endTime = c.getTime();
				this.process(startTime, endTime);
			}
			} catch (Exception e) {
				LogMessageUtils.build("实时往返价格处理", this.getClass().getSimpleName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
						insertNumber, taskTime, excuteStartTime);
				log.error(StringUtils.getStackTrace(e));
		}
	}

	/**
	 * 启动处理任务
	 * 
	 * @param start
	 *            格式： yyyyMMddHH
	 * @param end
	 *            格式： yyyyMMddHH
	 */
	public void process(final Date start, final Date end) {
		long startL = System.currentTimeMillis();
		log.info("LastestDoubleDataPreprocess start:"
				+ DateUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"));
		SourceEntity source = new SourceEntity();
		List<SourceEntity> listHome=source.findAll();
		List<String> strchannels= new ArrayList<String>();
		if(null!=listHome){
			for(SourceEntity entity:listHome){
				strchannels.add(entity.getHbaseKey());
			}
		}
	
		
		//重新初始化汇率
		ExchangeRateUtil.initExchangeRate(DateUtils.format(new Date(), "yyyyMMdd"));
		
		threadsNumber=strchannels.size();
		for (final String channel : strchannels) {
			
				taskExecutor.execute(new Runnable() {
					@Override
					public void run() {
						// log.info("start channel:"+channel+" area:"+area);
						Calendar cStart = Calendar.getInstance();
						cStart.setTime(start);
						Calendar cEnd = Calendar.getInstance();
						cEnd.setTime(end);
						while (cStart.before(cEnd)) {
							String startRow=RowkeyUtils.getRowKey(cStart.getTime(), channel,  RowkeyUtils.START_ROW_KEY);
							String endRow=RowkeyUtils.getRowKey(cStart.getTime(), channel,  RowkeyUtils.STOP_ROW_KEY);
							processDataByHour(startRow, endRow);// 20130319142000401
							cStart.add(Calendar.HOUR_OF_DAY, 1);// 开始时间加一小时
						}
						// log.info("end channel："+channel+" area："+area);
						
					}
				});
			
		}
		
		log.info("LastestDoubleDataPreprocess end ："
				+ DateUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS")
				+ " spend time:" + (System.currentTimeMillis() - startL));

	}

	/*-
	 * 
	 * 充分利用hbase数据库rowkey的字典序及rowkey的设计，将相同的航班放在相邻的位置进行统计，减少内存的占用
	 * 
	 * 按小时处理每小时的最低价
	 * 	 BASE_DATA表的 rowkey设计
	 *	 抓取基础数据表(BASE_DATA)：
	 *	 抓取日期(年月日时)_渠道_区域ID_起点(使用三字码)_终点(使用三字码)_航班ID_航班日期(年月日)_5位随即数
	 *	         10位 +    5位  +2位 +    3位   +        3位      + 8位+    7位 +    5位
	 * @param startRowKey
	 * @param endRowKey
	 * @throws IOException
	 * @throws Exception
	 */
	private void processDataByHour(String startRowKey,
			String endRowKey) {
		ResultScanner rs = null;
		List<DoublePlaneInfoEntity> list =new ArrayList<DoublePlaneInfoEntity>(512);
		try {
			try {
				rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
						startRowKey, endRowKey,new String[]{"planeinfo","returnPlaneInfosInfo"}, null);
				rs.iterator();
			} catch (Exception e) {
				if (rs != null)
					rs.close();
				rs = null;
				try {
					Thread.currentThread().sleep(10 * 1000);
				} catch (Exception e1) {
					e1.printStackTrace();
				}
				rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
						startRowKey, endRowKey,new String[]{"planeinfo","returnPlaneInfosInfo"}, null);
			}
			
			Map<String, Map<String, String>> flightFCV = new HashMap<String, Map<String, String>>();// 保存HBASE中存储的信息
			String rowId =null;
			
			for (Result res : rs) {
				try {// 某一条航班数据的错误，不应该影响其他航班数据的预处理
					//System.out.println(new String(res.getRow()));
				
					rowId=new String(res.getRow());
					tasksNumber++;
					flightFCV = HBaseHelper.transformHBaseRowToMap(res);
					DoublePlaneInfoEntity entity= this.mapToEntity(flightFCV,rowId);
					if(null==entity) continue;
					entity.setRowkey(rowId);
					list.add(entity);
					if(null!=list && list.size()>=500){
						this.saveToMysql(list);
						list.clear();
					}
					
				} catch (Exception e) {
					failuresNumber++;
					log.error(rowId);
					log.error(StringUtils.getStackTrace(e));
					FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.lastDouble, this.getClass().getSimpleName(),startRowKey.substring(5,15), e);
					List<String> keys =new ArrayList<String>();
					keys.add(rowId);
					vo.setKeys(keys);
					FetlErrorUtils.saveToHbase(vo);
				}
			}
			if(null!=list && list.size()>0){
				this.saveToMysql(list);
				list.clear();
			}
		} catch (Exception e1) {
			log.error(StringUtils.getStackTrace(e1));
			FetlErrorMessage vo=FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.lastDouble, this.getClass().getSimpleName(),startRowKey.substring(5,15), e1);
			vo.setStartRowKey(startRowKey);
			vo.setEndRowkey(endRowKey);
			FetlErrorUtils.saveToHbase(vo);
			
		} finally {
			if (rs != null)
				rs.close();
			rs = null;
			executeNumber++;
			if(executeNumber==threadsNumber){
				LogMessageUtils.build("实时往返价格处理", this.getClass().getSimpleName(), 2, null, 
						failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime);
			}
		}
	}

	private void saveToMysql(List<DoublePlaneInfoEntity> list) {
		List<DoublePlaneInfoEntity> oldList=null;
		List<DoublePlaneInfoEntity> deleteList=new ArrayList<DoublePlaneInfoEntity>();
		DoublePlaneInfoEntity entity =new DoublePlaneInfoEntity();
		for(DoublePlaneInfoEntity planInfo: list){
			if(null!=planInfo){
			try {
				oldList = planInfo.findByCondition(planInfo.getAttachHbaseKey(),
						planInfo.getFromCity(), planInfo.getToCity(),
						planInfo.getFlightDate(), planInfo.getReturnFlightDate(),
						planInfo.getFlightNo());
			
				if(null!=oldList && oldList.size()>0){
					for(DoublePlaneInfoEntity en:oldList){
						if(planInfo.getFlightDate().equals(en.getFlightDate())
								&& planInfo.getReturnFlightDate().equals(en.getReturnFlightDate())
								&& planInfo.getFlightNo().equals(en.getFlightNo())){
							deleteList.add(en);
						}
					}
				}
			} catch (SQLException e) {
				log.error(StringUtils.getStackTrace(e));
			}
		}
		}
		try {
			entity.delete(deleteList);
		} catch (SQLException e) {
			log.error(StringUtils.getStackTrace(e));
		}
		try {
			entity.create(list);
			insertNumber+=list.size();
		} catch (SQLException e) {
			failuresNumber+=list.size();
			log.error(StringUtils.getStackTrace(e));
		}
		
	}

	private DoublePlaneInfoEntity mapToEntity(Map<String, Map<String, String>> fcv,String rowId){
		DoublePlaneInfoEntity planInfo =null;
		double refePrice=ExchangeRateUtil.getRefePrice(fcv.get("planeinfo").get("currency"));

		planInfo= HbaseToPlaneInfoUtil.hbaseToPlaneInfo(fcv,refePrice,DoublePlaneInfoEntity.class);
		
		if(null==planInfo || null==planInfo.getAttachHbaseKey()||
			null==planInfo.getFromCity()|| null==planInfo.getToCity() ||
			null==planInfo.getFlightDate() || null==planInfo.getReturnFlightDate()||
			null== planInfo.getFlightNo()){
			if (null == planInfo) {
				//log.error("planeInfo is null");
			} else if (null == planInfo.getAttachHbaseKey()) {
				log.error("planInfo.getAttachHbaseKey() is null");
			} else if (null == planInfo.getFromCity()) {
				log.error("planInfo.getFromCity() is null");
			} else if (null == planInfo.getToCity()) {
				log.error("planInfo.getToCity() is null");
			} else if (null == planInfo.getFlightDate()) {
				log.error("planInfo.getFlightDate() is null");
			} else if (null == planInfo.getReturnFlightDate()) {
				log.error("planInfo.getReturnFlightDate() is null");
			} else if (null == planInfo.getFlightNo()) {
				log.error("planInfo.getFlightNo() is null");
			}
			
			return null;
		}

		return planInfo;
		
//		try {
//			if(null!=planInfo){
//				oldList = planInfo.findByCondition(planInfo.getAttachHbaseKey(),
//						planInfo.getFromCity(), planInfo.getToCity(),
//						planInfo.getFlightDate(), planInfo.getReturnFlightDate(),
//						planInfo.getFlightNo());
//				if(null!=oldList && oldList.size()>0){
//					for(DoublePlaneInfoEntity en:oldList){
//						if(planInfo.getFlightDate().equals(en.getFlightDate()) && planInfo.getReturnFlightDate().equals(en.getReturnFlightDate())
//								&& planInfo.getFlightNo().equals(en.getFlightNo())){
//							planInfo.delete(en);
//						}
//					}
//				}
//				planInfo.create(planInfo);
//				insertNumber++;
//			}
//		} catch (SQLException e) {
//			failuresNumber++;
//			log.error(rowId);
//			log.error(StringUtils.getStackTrace(e));
//		}
	}
	

	
	/**
	 * 最新的航班最低价（即以最新一次取到的数据为准）[官网、渠道]航班最低价LP_FLIGHT_DOUBLE rowkey：
	 * 渠道_起点(使用三字码)_终点(使用三字码)_航班日期(年月日)_返程日期(年月日)_(航班号)
	 * 
	 * @param planeMap
	 * @throws Exception
	 */
	private void saveLowestFlightPrice(Map<String, Map<String, String>> fcv) {

		Map<String, String> planeMap = fcv.get("planeinfo");
		String grabChannelNum = planeMap.get("grabChannelNum");
		String fromCity = planeMap.get("fromCity");
		String toCity = planeMap.get("toCity");
		if(planeMap.get("flightId")==null) return;
		if(planeMap.get("flightId").indexOf("|")>0){
			planeMap.put("flightId", planeMap.get("flightId").substring(0, planeMap.get("flightId").indexOf("|")));
		}
		if(planeMap.get("flightId").indexOf(",")>0){
			planeMap.put("flightId", planeMap.get("flightId").substring(0, planeMap.get("flightId").indexOf(",")));
		}
		String flightId = RowkeyUtils.getFlightIdHbaseKey(planeMap
				.get("flightId"));
		String flightBatchStartTime = planeMap.get("flightBatchStartTime")
				.substring(0, 8);
		
		String flightReturnTime=null;
		boolean isReturn=false;
		for(String infos : fcv.keySet()){
			if(infos.equalsIgnoreCase("returnPlaneInfosInfo")){
				isReturn=true;
				break;
			}
		}
		if(!isReturn){
			return;
		}
		Map<String,String> returnMap=fcv.get("returnPlaneInfosInfo");
		for(String key:returnMap.keySet()){
			if(key.startsWith("flightBatchStartTime")){
				flightReturnTime=returnMap.get(key).substring(0,8);
				break;
			}
			if(key.startsWith("flightBatchDate")){
				flightReturnTime=returnMap.get(key).replace("-", "").substring(0,8);
				break;
			}
		}
		
		StringBuffer transitFlightNo = new StringBuffer();
	
		if(fcv.containsKey("transitsInfo")){
			Map<String, String> transit =fcv.get("transitsInfo");

			for(String key : transit.keySet()){
				if(key.startsWith("flightNo")){
					String flightNo=transit.get(key);
					if(flightNo.indexOf("|")>0){
						flightNo=flightNo.substring(0, flightNo.indexOf("|"));
					}
					if(flightNo.indexOf(",")>0){
						flightNo=flightNo.substring(0, flightNo.indexOf(","));
					}
					if(flightId.equals(flightNo))
						continue;
					transitFlightNo.append(RowkeyUtils.getFlightIdHbaseKey(flightNo));
				}
			}
		}
		if(grabChannelNum==null || fromCity==null || toCity==null || flightReturnTime==null){
			return;
		}
		for (int j = transitFlightNo.length(); j < 16; j++) {
			transitFlightNo.append("0");
		}
		// 每小时的[官网、渠道]航班最低价LP_FLIGHT_DOUBLE rowkey：
		// 渠道_起点(使用三字码)_终点(使用三字码)_航班日期(年月日)_返程日期(年月日)_(航班号)_(中转航班号1)_(中转航班号2)
		StringBuffer sb = new StringBuffer();// 组装rowkey
		sb.append(grabChannelNum);
		sb.append(fromCity);
		sb.append(toCity);
		sb.append(flightBatchStartTime);
		sb.append(flightReturnTime);
		sb.append(flightId);
		sb.append(transitFlightNo);
	//	log.info("insert into hbase lp_flight_double:" + sb.toString());
		this.deleteTableRow("LP_FLIGHT_DOUBLE", sb.toString());
		for (Entry<String, Map<String, String>> fe : fcv.entrySet()) {
			Put put = new Put(sb.toString().getBytes());
			for (Entry<String, String> ce : fe.getValue().entrySet()) {
				put.add(fe.getKey().getBytes(), ce.getKey().getBytes(), ce
						.getValue().getBytes());
			}
			try {
				HBaseHelper.addRecord("LP_FLIGHT_DOUBLE", put);
			} catch (RetriesExhaustedWithDetailsException e) {
				failuresNumber++;
				e.printStackTrace();
				/*try {
					log.info("LP_FLIGHT_DOUBLE"+ sb.toString()+
							" HBase 可能正在进行分区操作，过2分钟再试！"
							+ StringUtils.getStackTrace(e));
					Thread.currentThread().sleep(120 * 1000);
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}*/
			} catch (Exception e) {
				failuresNumber++;
				log.info(StringUtils.getStackTrace(e));
			}
		}
		insertNumber++;
	}

	private void deleteTableRow(String tableName, String rowkey) {
		try {
			HBaseHelper.delRecord(tableName, rowkey);
		} catch (Exception e2) {
			// log.info("insert before delete rowkey="+rowkey+"   "+e2.getMessage());
		}
	}
	

}
